package com.zhuwei.kafka.streams;

import com.zhuwei.kafka.constant.Topic;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.state.KeyValueStore;

import java.util.Arrays;
import java.util.Properties;

/**
 * 官方文档：http://kafka.apache.org/documentation/streams/
 * 实现案例：实时处理topic first中的数据每条,按空格拆分
 * 实现案例流程: producer-->topic first-->(数据清理)-->topic three-->consumer
 * @author zhuiwei
 * @version 2019/05/14
 */
public class WordCountApplication {

    public static void main(String[] args) throws InterruptedException {
        /**
         * 设置参数
         * APPLICATION_ID_CONFIG:Streams应用程序必须要有一个应用ID,用于协调应用实例，也用于命名内部的本地存储和相关主题
         * 对于同一个kafka集群里的每一个Streams应用来说，这个名字可以随意起，但必须是唯一
         *
         * BOOTSTRAP_SERVERS_CONFIG：kafka服务器服务器和ip
         * DEFAULT_KEY_SERDE_CLASS_CONFIG：读写数据key序列化
         * DEFAULT_VALUE_SERDE_CLASS_CONFIG：读写数据value序列化
         */
        Properties prop = new Properties();
        prop.put(StreamsConfig.APPLICATION_ID_CONFIG, "mywordcount");
        prop.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.154.101:9092");
        prop.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        prop.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
		/*
		 * 通过KStreamBuilder得到KStream
		 *      builder.stream(String topic) 要处理的Topic作为参数,使用builder的stram方法得到一个KStream流
		 */
        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> textLines = builder.stream("first");
        /**
         * .flatMapValues：将topic中得到的流中的每一个值都变成小写然后以空格拆分成单词
         * .groupBy：将文本单词分组为消息键
         * .count：计算每个单词(消息键)的出现次数。
         * wordcounts.toStream().to(String topic, Produced<K, V> var2):将更新日志流转为记录流然后输出到另一个Topic
         */
        KTable<String, Long> wordcounts = textLines
                .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split(" ")))
                .groupBy((key, word) -> word)
                .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("counts-store"));
        wordcounts.toStream().to("three", Produced.with(Serdes.String(), Serdes.Long()));

        /**
         * new KafkaStreams()：基于拓扑和配置属性定义一个KafkaStreams对象
         * streams.start()：启动kafkastreams引擎
         * Thread.sleep(5000L)：一般情况下，Streams应用程序会一直运行下去，此处由于模拟测试，数据量少，我们就让线程休眠一段时间然后停止运行
         * streams.close()：停止运行
         */
        KafkaStreams streams = new KafkaStreams(builder.build(), prop);
        streams.start();
//		Thread.sleep(5000L);
//		streams.close();
    }
}
